package rabbitmq

import (
	"encoding/json"
	"fmt"
	"github.com/streadway/amqp"
	"mts/user_srv/global"
)

type RabbitMQ struct {
	conn *amqp.Connection
	ch   *amqp.Channel
	q    amqp.Queue
}

func NewRabbitMQ() (*RabbitMQ, error) {
	mqInfo := global.ServerConfig.RabbitMQInfo
	conn, err := amqp.Dial(fmt.Sprintf("amqp://%s:%s@%s:%d", mqInfo.User, mqInfo.Pass, mqInfo.Host, mqInfo.Port))

	if err != nil {
		return nil, err
	}

	ch, err := conn.Channel()
	if err != nil {
		conn.Close()
		return nil, err
	}
	args := amqp.Table{
		"x-dead-letter-exchange":    "dlx_logs_exchange",
		"x-dead-letter-routing-key": "dlq.logs.key",
		"x-message-ttl":             60000,       // 消息超时时间
		"x-overflow":                "drop-head", // 队列满时丢弃旧消息
	}

	q, err := ch.QueueDeclare(
		"logs_queue", // name
		true,         // durable
		false,        // delete when unused
		false,        // exclusive
		false,        // no-wait
		args,         // arguments
	)
	if err != nil {
		ch.Close()
		conn.Close()
		return nil, err
	}

	return &RabbitMQ{
		conn: conn,
		ch:   ch,
		q:    q,
	}, nil
}

func (rmq *RabbitMQ) Close() error {
	err := rmq.ch.Close()
	if err != nil {
		return err
	}
	return rmq.conn.Close()
}

func (rmq *RabbitMQ) Publish(logData map[string]interface{}) error {
	body, err := json.Marshal(logData)
	if err != nil {
		return err
	}
	if err = rmq.ch.Tx(); err != nil {
		return fmt.Errorf("事务开启失败:%s", err.Error())
	}
	err = rmq.ch.Publish(
		"",
		rmq.q.Name,
		false,
		false,
		amqp.Publishing{
			ContentType:  "application/json",
			Body:         body,
			Expiration:   "600000",        //超时时间
			DeliveryMode: amqp.Persistent, // 持久化消息
		})
	if err != nil {
		if err = rmq.ch.TxRollback(); err != nil {
			return fmt.Errorf("事务回滚失败:%s", err.Error())
		}
		return err
	}
	if err = rmq.ch.TxCommit(); err != nil {
		return fmt.Errorf("提交事务失败:%s", err.Error())
	}
	return nil
}

//func (rmq *RabbitMQ) PublishWithConfirm(logData map[string]interface{}) error {
//	body, err := json.Marshal(logData)
//	if err != nil {
//		return err
//	}
//	// 开启 Confirm 模式
//	err = rmq.ch.Confirm(false)
//	if err != nil {
//		return fmt.Errorf("failed to enable confirm mode: %v", err)
//	}
//
//	// 创建用于接收确认的 channel
//	ackChan := make(chan amqp.Confirmation)
//	rmq.ch.NotifyPublish(ackChan)
//
//	// 发布消息
//	err = rmq.ch.Publish(
//		"",         // exchange
//		rmq.q.Name, // routing key
//		false,      // mandatory
//		false,      // immediate
//		amqp.Publishing{
//			ContentType: "application/json",
//			Body:        body,
//		})
//	if err != nil {
//		return fmt.Errorf("failed to publish message: %v", err)
//	}
//
//	// 等待确认结果
//	select {
//	case confirm := <-ackChan:
//		if confirm.Ack {
//			fmt.Println("Message confirmed by broker")
//		} else {
//			return fmt.Errorf("message was nacked by broker")
//		}
//	}
//
//	return nil
//}
